package com.jacquesh;

import com.jacquesh.model.User;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlinkDemo0003 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment=
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream= environment
                .socketTextStream("192.168.88.180",8888);
        DataStream<User> newStream= stream
                .flatMap(new FlatMapFunction<String, User>() {
                    @Override
                    public void flatMap(String s, Collector<User> collector)  {
                        try {
                            String[] sp=s.split(",");
                            collector.collect(User.builder()
                                    .id(Long.parseLong(sp[0]))
                                    .name(sp[1])
                                    .age(Integer.parseInt(sp[2]))
                                    .build());
                        }catch (Exception ex){
                            System.out.println(ex);
                        }
                    }
                })
                .filter(new RichFilterFunction<User>() {
                    @Override
                    public boolean filter(User user) throws Exception {
                        /**只处理age 大于25的参数*/
                        if (user.getAge()>25){
                            return true;
                        }
                        return false;
                    }
                })
                /**依**/
                .keyBy(User::getName)
                .sum("age");

        newStream.print(">>>>>sout>>>>>");
        environment.execute("KeyBy demo");

    }
}
